package cn.zx.rmq.config;

import cn.zx.rmq.listener.MQConsumeMsgListenerProcessor;
import cn.zx.rmq.listener.MQConsumeMsgOrderListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;

/**
 * @Package: cn.zx.rmq.config
 * @ClassName: MQOrderConsumerConfiguration
 * @Author: Think
 * @CreateTime: 2021/6/30 10:31
 * @Description:
 */
@Slf4j
@SpringBootConfiguration
public class MQOrderConsumerConfiguration {
    @Value("${rocketmq.consumer.order.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.order.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.order.consumeThreadMin}")
    private int consumeThreadMin;
    @Value("${rocketmq.consumer.order.consumeThreadMax}")
    private int consumeThreadMax;
    @Value("${rocketmq.consumer.order.topics}")
    private String topics;
    @Value("${rocketmq.consumer.order.consumeMessageBatchMaxSize}")
    private int consumeMessageBatchMaxSize;
    @Autowired
    private MQConsumeMsgOrderListener mqConsumeMsgOrderListener;

    @Bean(name = "orderConsumer")
    public DefaultMQPushConsumer getRocketMQConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(mqConsumeMsgOrderListener);
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 设置一次消费消息的条数，默认为1条
         */
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        /**
         * 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，则tag使用*；如果需要指定订阅该主题下的某些tag，则使用||分割，例如tag1||tag2||tag3
         */
        String[] topicTagsArr = topics.split(";");
        for (String topicTags : topicTagsArr) {
            String[] topicTag = topicTags.split("~");
            consumer.subscribe(topicTag[0],topicTag[1]);
        }
        consumer.start();
        return consumer;
    }
}
